package com.red.wood.task;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.red.wood.util.JarToolUtil;
import com.red.wood.util.SysProps;

public class HiveToDbTask {
	
	private static Log log = LogFactory.getLog(HiveToDbTask.class);
	
	/**
	 * 通过sqoop，将hive表中数据导入到db中
	 * @param table
	 * @param ds
	 */
	public static void exeHiveSqoop(String table, String ds) {
		
		log.info("=========sqoop【"+table+"】导入任务开始============");
		
		Process process = null;
		BufferedReader br = null;
		try {
			SysProps sysProps = SysProps.getInstance();

			String hive_databse = sysProps.getValue("hive.database");
			String url = sysProps.getValue("jdbc.isdms.oracle.url");
			String username = sysProps.getValue("jdbc.isdms.oracle.username");
			String password = sysProps.getValue("jdbc.isdms.oracle.password");

			String path = JarToolUtil.getJarParentDir() + File.separator + "conf" + File.separator+"sqoop_db.sh";
			
			ProcessBuilder pb = new ProcessBuilder(
					"sh", //执行的命令
					path, //sqoop脚本
					hive_databse, //hive数据库
					url, //数据库URL
					username, //数据库用户名,转换成大写 ,防止导入的时sqoop报错
					password, //数据密码
					table, //hive表与oracle一样，此处必须小写
					ds //日期分区
			);
			// 是否合并标准错误和标准输出
			pb.redirectErrorStream(true);
			//pb.directory(new File("D:"+File.separator+"sqoop_db.sh"));//脚本文件路径
			process = pb.start();
			// 执行命令
			br = new BufferedReader(new InputStreamReader(process.getInputStream(), "UTF-8"));
			String s = "";
			while ((s = br.readLine()) != null) {
				// 打印信息
				log.info(s);
			}
			log.info("=========sqoop【"+table+"】导入任务结束============");
		} catch (IOException e) {
			log.error("-------通过sqoop，将hive中的数据导入到DB表中IO异常错误--------：", e);
		} finally {
			if (br != null) {
				try {
					br.close();
				} catch (IOException e) {
					log.error(e.getMessage(), e);
				}
			}
			// 显式将process.destroy()掉，防止消耗资源
			if (process != null) {
				process.destroy();
			}
		}
	}
}